package com.lagou.common;

import com.lagou.zookeeper.NodeChangeListener;
import com.lagou.zookeeper.RPCRegistryHandler;
import com.lagou.zookeeper.config.ConfigKeeper;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ZkRegistryHandler implements RPCRegistryHandler {

    private static final String ZK_PATH_SPLITER = "/";
    private static final String RPC_ZK_ROOT = ZK_PATH_SPLITER + "rpc-provider" + ZK_PATH_SPLITER;
    private List<NodeChangeListener> listenerList = new ArrayList<>();
    private final String url;
    private final CuratorFramework client;

    private List<String> serviceList;
    private static final ScheduledExecutorService REPORT_WORKER = Executors.newScheduledThreadPool(5);

    /**
     * 连接注册中心，服务端注册监听
     * @param zkPath
     */
    public ZkRegistryHandler(final String zkPath){
        url = zkPath;
        this.client = CuratorFrameworkFactory.builder()
                .connectString(zkPath)
                .retryPolicy(new ExponentialBackoffRetry(1000,3))
                .build();

        client.getConnectionStateListenable().addListener((curatorFramework, connectionState) -> {
            if(ConnectionState.CONNECTED.equals(connectionState)){
                System.out.println("注册中心连接成功");
            }
        });

        client.start();

        // 定时上报
        final ConfigKeeper configKeeper = ConfigKeeper.getInstance();
        final boolean providerSide = configKeeper.isProviderSide();
        final int interval = configKeeper.getInterval();
        if(!providerSide && interval > 0){
            REPORT_WORKER.scheduleWithFixedDelay(() -> System.out.println("触发定时任务"),interval,interval, TimeUnit.SECONDS);
        }
    }


    @Override
    public boolean registry(final String serviceName, final String nettyHost, final int nettyPort) {
        // 常量 全路径 固定名称
        String zkPath = providePath(serviceName);

        if(!exists(zkPath)){
            // 创建持久节点
            create(zkPath,false);
        }

        String instancePath = zkPath + ZK_PATH_SPLITER + nettyHost + ":" + nettyPort;
        // 创建临时节点
        create(instancePath, true);
        return true;
    }

    /**
     * 创建节点
     * @param zkPath
     * @param ephemeral 创建模式 临时节点和持久节点
     */
    private void create(final String zkPath, final boolean ephemeral) {
        CreateMode createMode;
        if (ephemeral) {
            createMode = CreateMode.EPHEMERAL;
        }else{
            createMode = CreateMode.PERSISTENT;
        }

        try {
            client.create().creatingParentContainersIfNeeded().withMode(createMode).forPath(zkPath);
        }catch (KeeperException.NodeExistsException ex){
            System.out.println("路径已存在" + zkPath);
        }
        catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException();
        }
    }

    private boolean exists(String zkPath) {
        try {
            if(client.checkExists().forPath(zkPath) != null){
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 注册监听及监听到的方法执行
     * @param serviceName
     * @param zkPath
     */
    private void registryWatch(String serviceName, String zkPath) {
        PathChildrenCache nodeCache = new PathChildrenCache(client, zkPath, true);
        try {
            nodeCache.getListenable().addListener((client, pathChildrenCacheEvent) -> {
                serviceList = client.getChildren().forPath(zkPath);
                listenerList.forEach(nodeChangeListener -> {
                    System.out.println("节点发生变化");
                    nodeChangeListener.notify(serviceName, serviceList, pathChildrenCacheEvent);
                });
            });
            nodeCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    private String providePath(String serviceName) {
        return RPC_ZK_ROOT + serviceName + ZK_PATH_SPLITER + "provider";
    }

    @Override
    public List<String> discovery(final String serviceName) {
        // 完成服务地址拼接
        final String path = providePath(serviceName);

        // 第一次为空，需要从Zk获取，后续通过watcher更新
        if (serviceList == null || serviceList.isEmpty()) {
            System.out.println("首次探查该地址");
            try {
                serviceList = client.getChildren().forPath(path);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        this.registryWatch(serviceName,path);
        return serviceList;
    }

    @Override
    public void addListener(NodeChangeListener listener) {
        listenerList.add(listener);
    }

    @Override
    public void destroy() {
        client.close();
    }

    @Override
    public void notify(String children, List<String> serviceList, PathChildrenCacheEvent pathChildrenCacheEvent) {
        for (NodeChangeListener nodeChangeListener : listenerList) {
            nodeChangeListener.notify(children, serviceList, pathChildrenCacheEvent);
        }
    }
}
